package com.etc;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.beanutils.BeanUtilsBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.codehaus.jackson.map.util.BeanUtil;

import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;


/**
 * @Auther: zhezhe
 * @Date: 2018/10/3010:24
 * @Description:
 */
public class MapReduceObjectSort {
    public static class MapObjectSort extends Mapper<LongWritable, Text,Text,ObjectSort>{
        /**
         * 转换编码格式
         * @param text
         * @param encoding
         * @return
         */
        public static Text transformTextToUTF8(Text text, String encoding) {
            String value = null;
            try {
                value = new String(text.getBytes(), 0, text.getLength(), encoding);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
            return new Text(value);
        }


        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
           //将value转为GBK编码格式的
            Text gbk = transformTextToUTF8(value, "GBK");
            String line = gbk.toString();
            String[] s = line.split(",");
            String orderNum = s[0];
            String user = s[1];
            String goodsName = s[2];
            double goodsPrice = Double.parseDouble(s[3]);
            int goodsNum = Integer.parseInt(s[4]);

            context.write(new Text(orderNum), new ObjectSort(orderNum,user,goodsName,goodsPrice,goodsNum));
        }
    }


    public static class ReduceObjectSort extends Reducer<Text,ObjectSort,ObjectSort, NullWritable>{

        List<ObjectSort> objectList = new ArrayList<ObjectSort>();

        @Override
        protected void reduce(Text key, Iterable<ObjectSort> values, Context context) throws IOException, InterruptedException {

            for (ObjectSort value : values) {
                //因为list添加的是具体的对象，所以要先将对象new出来
                ObjectSort o = new ObjectSort();
                try {
                    //将value中的数据拷贝到对象o中
                    BeanUtils.copyProperties(o,value);
                } catch (IllegalAccessException e) {
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
                //往objectList中添加对象
                objectList.add(o);
            }
            //排序
            Collections.sort(objectList);

            for (int i = 0; i <objectList.size() ; i++) {
                if (i==3){
                    break;
                }
                context.write(objectList.get(i),NullWritable.get());
            }

        }

    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(MapReduceObjectSort.class);

        job.setMapperClass(MapObjectSort.class);
        job.setReducerClass(ReduceObjectSort.class);


        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(ObjectSort.class);


        job.setOutputKeyClass(ObjectSort.class);
        job.setOutputValueClass(NullWritable.class);

        File file = new File("E:\\Mapreduce案例\\MapObjectSort\\练习数据\\output");
        if (file.exists()){
            FileUtil.fullyDelete(file);
        }

        FileInputFormat.addInputPath(job,new Path("E:\\Mapreduce案例\\MapObjectSort\\练习数据\\input\\*"));
        FileOutputFormat.setOutputPath(job,new Path("E:\\Mapreduce案例\\MapObjectSort\\练习数据\\output"));


        job.setNumReduceTasks(1);

        System.exit(job.waitForCompletion(true) ? 0 :1 );

    }
}
